package org.eleorc.flow;

import java.util.concurrent.Flow;
import java.util.concurrent.atomic.AtomicInteger;

public class FlowDemo01 {

    public static void main(String[] args) throws InterruptedException {
        NumberPublisher publisher = new NumberPublisher();
        Flow.Subscriber<Integer> subscriber = new PrintSubscriber();
        publisher.subscribe(subscriber);

        Thread.sleep(3000);
    }

    public static class NumberPublisher implements Flow.Publisher<Integer>{

        private AtomicInteger count = new AtomicInteger(1);

        @Override
        public void subscribe(Flow.Subscriber<? super Integer> subscriber) {
            subscriber.onSubscribe(new Flow.Subscription() {
                @Override
                public void request(long n) {
                    for (long i = 0; i < n && count.get() <= 10; i++) {
                        int number = count.getAndIncrement();
                        subscriber.onNext(number);
                    }
                    if (count.get() > 10) {
                        subscriber.onComplete();
                    }
                }

                @Override
                public void cancel() {

                }
            });
        }
    }

    public static class PrintSubscriber implements Flow.Subscriber<Integer> {

        private Flow.Subscription subscription;
        private boolean isCompleted = false;
        @Override
        public void onSubscribe(Flow.Subscription subscription) {
            this.subscription = subscription;
            this.subscription.request(1); // Request the first item.
        }

        @Override
        public void onNext(Integer item) {
            System.out.println(Thread.currentThread().getName()+" Received: " + item);
            if (!isCompleted) {
                subscription.request(1); // Request the next item after processing the current one.
            }
        }

        @Override
        public void onError(Throwable throwable) {
            throwable.printStackTrace();
        }

        @Override
        public void onComplete() {
            if (!isCompleted) {
                System.out.println(Thread.currentThread().getName()+" Processing complete.");
                isCompleted = true;
            }
//            subscription.cancel();
//            System.out.println(Thread.currentThread().getName()+" Subscription cancelled.");
        }
    }
}
